package com.lagou.rpc.consumer.client;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

@Component
public class ConsumerReporter {
    private ReentrantLock lock = new ReentrantLock();
    private Map<ServerInfo, ConsumeStatics> serverStatics = new HashMap<>();

    @Autowired
    private Registry registry;

    @Autowired
    private ClientConfig clientConfig;

    private ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(1);

    // 提供给客户端调用
    public void report(ServerInfo serverInfo, ConsumeStatics statics) {
        lock.lock();

        try {
            serverStatics.put(serverInfo, statics);
        }
        finally {
            lock.unlock();
        }
    }

    @PostConstruct
    public void init() {
        pool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("===Start report===");
                    updateServerStatics();
                }
                catch (Exception ignored) { }
            }
        }, 90, 90, TimeUnit.SECONDS);
    }

    private void updateServerStatics() {
        // 轮询算法不上传统计信息
        if (ClientConfig.ALG_ROUND_ROBIN.equals(clientConfig.getAlgorithm())) {
            return;
        }

        lock.lock();

        long currTimeMills = System.currentTimeMillis();
        try {
            for (Map.Entry<ServerInfo, ConsumeStatics> entry : serverStatics.entrySet()) {
                ServerInfo serverInfo = entry.getKey();
                ConsumeStatics statics = entry.getValue();

                if (currTimeMills - statics.getLastConsumeTime() > 1000 * 3000) {
                    registry.clearServerStatics(serverInfo);
                }
                else {
                    registry.updateServerStatics(serverInfo, statics);
                }
            }
        }
        finally {
            lock.unlock();
        }
    }

}
